--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit f94fe0ec088a5d468205b91259ef5e7e94684050
Parents : 730ab28
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-16T09:33:19-06:00
Improve integrity management with advanced checks and metadata support
- Introduced entropy calculations to detect content type shifts in files.
- Added SQLite integrity checks to verify database structure and prevent tampering.
- Updated the manifest saving process to include metadata for files, such as entropy and size.
- Expanded ignored file patterns for volatile LXMF components.
- Implemented extensive unit tests to validate new features and ensure robustness.
Changes
3 files changed, 324 insertions(+), 32 deletions(-)
Diff
diff --git a/meshchatx/src/backend/integrity_manager.py b/meshchatx/src/backend/integrity_manager.py
index a6a71905..148eb382 100644
--- a/meshchatx/src/backend/integrity_manager.py
+++ b/meshchatx/src/backend/integrity_manager.py
@@ -1,7 +1,9 @@
import fnmatch
import hashlib
import json
+import math
import os
+import sqlite3
from datetime import UTC, datetime
from pathlib import Path
@@ -22,6 +24,7 @@ class IntegrityManager:
".DS_Store",
"Thumbs.db",
"integrity-manifest.json",
+ "outbound_stamp_costs",
]
def __init__(self, storage_dir, database_path, identity_hash=None):
@@ -40,11 +43,17 @@ class IntegrityManager:
# We only ignore these if they are inside the lxmf_router directory
# to avoid accidentally ignoring important files with similar names.
if "lxmf_router" in path_parts:
+ # Added more volatile LXMF patterns
if any(
- part in ["announces", "storage", "identities"] for part in path_parts
+ part in ["announces", "storage", "identities", "tmp"]
+ for part in path_parts
):
return True
+ # Specifically ignore stamp costs which are frequently updated
+ if path_parts[-1] == "outbound_stamp_costs":
+ return True
+
# Check for other generally ignored directories
if any(
part in ["tmp", "recordings", "greetings", "docs", "bots", "ringtones"]
@@ -69,8 +78,48 @@ class IntegrityManager:
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
+ def _calculate_entropy(self, file_path):
+ """Calculate Shannon entropy of a file to detect content type shifts."""
+ if not os.path.exists(file_path):
+ return 0
+ try:
+ with open(file_path, "rb") as f:
+ # Sample up to 64KB for performance
+ data = f.read(65536)
+ if not data:
+ return 0
+
+ byte_counts = [0] * 256
+ for b in data:
+ byte_counts[b] += 1
+
+ entropy = 0
+ total = len(data)
+ for count in byte_counts:
+ if count > 0:
+ p = count / total
+ entropy -= p * math.log2(p)
+ return entropy
+ except Exception:
+ return 0
+
+ def _check_db_integrity(self, db_path):
+ """Use SQLite's PRAGMA integrity_check to verify the database."""
+ if not os.path.exists(db_path):
+ return False, "Database file does not exist"
+ try:
+ # Use read-only mode for checking
+ conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+ cursor = conn.cursor()
+ cursor.execute("PRAGMA integrity_check")
+ result = cursor.fetchone()[0]
+ conn.close()
+ return result == "ok", result
+ except Exception as e:
+ return False, str(e)
+
def check_integrity(self):
- """Verify the current state against the last saved manifest."""
+ """Verify the current state against the last saved manifest using advanced analytics."""
if not self.manifest_path.exists():
return True, ["Initial run - no manifest yet"]
@@ -80,13 +129,39 @@ class IntegrityManager:
issues = []
manifest_files = manifest.get("files", {})
+ manifest_metadata = manifest.get("metadata", {})
+ m_id = manifest.get("identity", "Unknown")
+
+ # Always check for identity mismatch first as it's a fundamental security issue
+ if self.identity_hash and m_id != "Unknown" and self.identity_hash != m_id:
+ issues.append(f"Identity mismatch! Manifest belongs to: {m_id}")
- # Check Database
+ # Check Database (Math-based structural check + Entropy stability + Hash)
if self.database_path.exists():
db_rel = str(self.database_path.relative_to(self.storage_dir))
actual_db_hash = self._hash_file(self.database_path)
- if actual_db_hash and actual_db_hash != manifest_files.get(db_rel):
- issues.append(f"Database modified: {db_rel}")
+
+ if actual_db_hash != manifest_files.get(db_rel):
+ # Check internal SQL integrity to see if it's just a dirty shutdown or actual tampering
+ is_db_ok, db_msg = self._check_db_integrity(self.database_path)
+ if not is_db_ok:
+ issues.append(f"Database structural issue: {db_msg}")
+ else:
+ # Check entropy stability to see if content type shifted significantly
+ actual_entropy = self._calculate_entropy(self.database_path)
+ saved_entropy = manifest_metadata.get(db_rel, {}).get("entropy")
+
+ if (
+ saved_entropy is not None
+ and abs(actual_entropy - saved_entropy) > 1.0
+ ):
+ issues.append(
+ f"Database structural anomaly (Entropy Δ: {abs(actual_entropy - saved_entropy):.2f})"
+ )
+ else:
+ issues.append(
+ f"Database binary signature mismatch: {db_rel}"
+ )
# Check other critical files in storage_dir
for root, _, files_in_dir in os.walk(self.storage_dir):
@@ -97,7 +172,7 @@ class IntegrityManager:
if self._should_ignore(rel_path):
continue
- # Database already checked separately, skip here to avoid double reporting
+ # Database handled separately
if full_path == self.database_path:
continue
@@ -105,13 +180,40 @@ class IntegrityManager:
if rel_path in manifest_files:
if actual_hash != manifest_files[rel_path]:
- issues.append(f"File modified: {rel_path}")
+ actual_entropy = self._calculate_entropy(full_path)
+ saved_entropy = manifest_metadata.get(rel_path, {}).get(
+ "entropy"
+ )
+ saved_size = manifest_metadata.get(rel_path, {}).get(
+ "size", 0
+ )
+ actual_size = full_path.stat().st_size
+
+ is_critical = any(
+ c in rel_path for c in ["identity", "config"]
+ )
+
+ if is_critical:
+ issues.append(
+ f"Critical security component integrity compromised: {rel_path}"
+ )
+ elif (
+ saved_entropy is not None
+ and abs(actual_entropy - saved_entropy) > 1.5
+ ):
+ issues.append(
+ f"Non-linear content shift detected in {rel_path} (Entropy Δ: {abs(actual_entropy - saved_entropy):.2f})"
+ )
+ elif saved_size and actual_size != saved_size:
+ issues.append(
+ f"File size divergence: {rel_path} ({saved_size} -> {actual_size} bytes)"
+ )
+ else:
+ issues.append(f"File signature mismatch: {rel_path}")
else:
- # New files are also a concern for integrity
- # but we only report them if they are not in ignored dirs/patterns
issues.append(f"New file detected: {rel_path}")
- # Check for missing files that were in manifest
+ # Check for missing files
for rel_path in manifest_files:
if self._should_ignore(rel_path):
continue
@@ -123,34 +225,22 @@ class IntegrityManager:
if issues:
m_date = manifest.get("date", "Unknown")
m_time = manifest.get("time", "Unknown")
- m_id = manifest.get("identity", "Unknown")
issues.insert(
0,
f"Last integrity snapshot: {m_date} {m_time} (Identity: {m_id})",
)
- # Check if identity matches
- if (
- self.identity_hash
- and m_id != "Unknown"
- and self.identity_hash != m_id
- ):
- issues.append(f"Identity mismatch! Manifest belongs to: {m_id}")
-
self.issues = issues
return len(issues) == 0, issues
except Exception as e:
- import traceback
-
- traceback.print_exc()
return False, [f"Integrity check failed: {e!s}"]
def save_manifest(self):
- """Snapshot the current state of critical files."""
+ """Snapshot the current state with extended mathematical metadata."""
try:
files = {}
+ metadata = {}
- # Hash all critical files in storage_dir recursively
for root, _, files_in_dir in os.walk(self.storage_dir):
for file in files_in_dir:
full_path = Path(root) / file
@@ -160,15 +250,20 @@ class IntegrityManager:
continue
files[rel_path] = self._hash_file(full_path)
+ metadata[rel_path] = {
+ "entropy": self._calculate_entropy(full_path),
+ "size": full_path.stat().st_size,
+ }
now = datetime.now(UTC)
manifest = {
- "version": 1,
+ "version": 2,
"timestamp": now.timestamp(),
"date": now.strftime("%Y-%m-%d"),
"time": now.strftime("%H:%M:%S"),
"identity": self.identity_hash,
"files": files,
+ "metadata": metadata,
}
with open(self.manifest_path, "w") as f:
diff --git a/tests/backend/test_integrity.py b/tests/backend/test_integrity.py
index ed53738b..da9748ed 100644
--- a/tests/backend/test_integrity.py
+++ b/tests/backend/test_integrity.py
@@ -1,6 +1,8 @@
import shutil
import tempfile
import unittest
+import os
+import sqlite3
from pathlib import Path
from meshchatx.src.backend.integrity_manager import IntegrityManager
@@ -13,9 +15,10 @@ class TestIntegrityManager(unittest.TestCase):
self.identities_dir = self.test_dir / "identities"
self.identities_dir.mkdir()
- # Create a dummy database
- with open(self.db_path, "w") as f:
- f.write("dummy db content")
+ # Create a real SQLite database
+ conn = sqlite3.connect(self.db_path)
+ conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)")
+ conn.close()
# Create a dummy identity
self.id_path = self.identities_dir / "test_id"
@@ -45,13 +48,13 @@ class TestIntegrityManager(unittest.TestCase):
"""Test detection of database modification."""
self.manager.save_manifest()
- # Modify DB
+ # Modify DB in a way that breaks SQLite integrity or at least changes hash
with open(self.db_path, "a") as f:
f.write("tampered")
is_ok, issues = self.manager.check_integrity()
self.assertFalse(is_ok)
- self.assertTrue(any("Database modified" in i for i in issues))
+ self.assertTrue(any("Database" in i for i in issues))
self.assertTrue(any("Last integrity snapshot" in i for i in issues))
def test_identity_mismatch(self):
@@ -64,7 +67,7 @@ class TestIntegrityManager(unittest.TestCase):
# Tamper a file to trigger issues list which includes the metadata check
with open(self.db_path, "a") as f:
- f.write("tampered")
+ f.write("more content to change hash")
is_ok, issues = self.manager.check_integrity()
self.assertFalse(is_ok)
@@ -81,7 +84,12 @@ class TestIntegrityManager(unittest.TestCase):
is_ok, issues = self.manager.check_integrity()
self.assertFalse(is_ok)
- self.assertTrue(any("File modified" in i for i in issues))
+ self.assertTrue(
+ any(
+ "Critical security component" in i or "File signature mismatch" in i
+ for i in issues
+ )
+ )
def test_new_identity_detected(self):
"""Test detection of unauthorized new identity files."""
diff --git a/tests/backend/test_integrity_extensive.py b/tests/backend/test_integrity_extensive.py
new file mode 100644
index 00000000..f9c17d9d
--- /dev/null
+++ b/tests/backend/test_integrity_extensive.py
@@ -0,0 +1,189 @@
+import shutil
+import tempfile
+import unittest
+import os
+import sqlite3
+import math
+import json
+from pathlib import Path
+from hypothesis import given, strategies as st, settings, HealthCheck
+
+from meshchatx.src.backend.integrity_manager import IntegrityManager
+
+
+class TestIntegrityManagerExtensive(unittest.TestCase):
+ def setUp(self):
+ self.test_dir = Path(tempfile.mkdtemp())
+ self.db_path = self.test_dir / "database.db"
+ self.storage_dir = self.test_dir / "storage"
+ self.storage_dir.mkdir()
+
+ # Create a valid SQLite database
+ conn = sqlite3.connect(self.db_path)
+ conn.execute("CREATE TABLE data (id INTEGER PRIMARY KEY, val TEXT)")
+ conn.execute("INSERT INTO data (val) VALUES ('initial')")
+ conn.commit()
+ conn.close()
+
+ self.manager = IntegrityManager(self.test_dir, self.db_path)
+
+ def tearDown(self):
+ shutil.rmtree(self.test_dir)
+
+ def test_entropy_mathematical_bounds(self):
+ """Verify entropy stays within [0, 8] for any byte sequence."""
+ # Test empty
+ empty_file = self.test_dir / "empty"
+ empty_file.touch()
+ self.assertEqual(self.manager._calculate_entropy(empty_file), 0)
+
+ # Test single byte repeated (minimum entropy)
+ zero_entropy_file = self.test_dir / "zero_entropy"
+ with open(zero_entropy_file, "wb") as f:
+ f.write(b"AAAAAAAA" * 100)
+ self.assertEqual(self.manager._calculate_entropy(zero_entropy_file), 0)
+
+ # Test all 256 bytes (maximum entropy)
+ max_entropy_file = self.test_dir / "max_entropy"
+ with open(max_entropy_file, "wb") as f:
+ f.write(bytes(range(256)))
+ # log2(256) = 8
+ self.assertAlmostEqual(
+ self.manager._calculate_entropy(max_entropy_file), 8.0, places=5
+ )
+
+ @settings(suppress_health_check=[HealthCheck.too_slow], deadline=None)
+ @given(st.binary(min_size=1, max_size=1024))
+ def test_entropy_property(self, data):
+ """Property: Entropy is always between 0 and 8 for non-empty data."""
+ temp_file = self.test_dir / "prop_test"
+ with open(temp_file, "wb") as f:
+ f.write(data)
+
+ entropy = self.manager._calculate_entropy(temp_file)
+ self.assertGreaterEqual(entropy, 0)
+ self.assertLessEqual(entropy, 8.000000000000002) # Float precision
+
+ def test_db_structural_tamper_detection(self):
+ """Simulate actual SQLite corruption that bypasses hash-only checks."""
+ self.manager.save_manifest()
+
+ # Corrupt the database file header or internal structure
+ # Overwriting the first few bytes (SQLite header) is a guaranteed fail
+ with open(self.db_path, "r+b") as f:
+ f.seek(0)
+ f.write(b"NOTASQLITEFILE")
+
+ is_ok, issues = self.manager.check_integrity()
+ self.assertFalse(is_ok, f"Integrity should fail. Issues: {issues}")
+ self.assertTrue(
+ any(
+ "Database structural issue" in i or "Database structural anomaly" in i
+ for i in issues
+ ),
+ f"Expected structural issue in: {issues}",
+ )
+
+ def test_entropy_shift_detection(self):
+ """Test detection of content type change (e.g. replacing text with random bytes)."""
+ # 1. Start with a highly structured file (low entropy)
+ # Use a non-critical filename to trigger the entropy check branch
+ data_file = self.test_dir / "user_data.bin"
+ with open(data_file, "wb") as f:
+ f.write(b"A" * 5000)
+
+ self.manager.save_manifest()
+
+ # 2. Replace with high-entropy data (random bytes)
+ with open(data_file, "wb") as f:
+ f.write(os.urandom(5000))
+
+ is_ok, issues = self.manager.check_integrity()
+ self.assertFalse(is_ok, f"Integrity should fail. Issues: {issues}")
+ self.assertTrue(
+ any("Non-linear content shift" in i or "Entropy Δ" in i for i in issues),
+ f"Expected entropy shift in: {issues}",
+ )
+
+ def test_ignore_patterns_extensive(self):
+ """Verify all volatile LXMF/RNS patterns are correctly filtered."""
+ volatile_files = [
+ "lxmf_router/lxmf/outbound_stamp_costs",
+ "lxmf_router/storage/some_volatile_file",
+ "lxmf_router/announces/ann_data",
+ "lxmf_router/tmp/uploading",
+ "database.db-wal",
+ "database.db-shm",
+ "something.tmp",
+ ".DS_Store",
+ ]
+
+ for v in volatile_files:
+ rel_path = Path(v)
+ full_path = self.test_dir / rel_path
+ full_path.parent.mkdir(parents=True, exist_ok=True)
+ full_path.touch()
+ self.assertTrue(
+ self.manager._should_ignore(str(rel_path)), f"Failed to ignore {v}"
+ )
+
+ def test_critical_file_protection(self):
+ """Ensure identity and config changes are always treated as critical."""
+ id_file = self.test_dir / "identity"
+ id_file.write_text("secure_key")
+
+ self.manager.save_manifest()
+
+ # Minor modification (stays low entropy)
+ id_file.write_text("secure_kez")
+
+ is_ok, issues = self.manager.check_integrity()
+ self.assertFalse(is_ok)
+ self.assertTrue(any("Critical security component" in i for i in issues))
+
+ def test_missing_file_detection(self):
+ """Verify missing files are detected even if not critical."""
+ misc_file = self.test_dir / "misc.txt"
+ misc_file.write_text("data")
+
+ self.manager.save_manifest()
+ misc_file.unlink()
+
+ is_ok, issues = self.manager.check_integrity()
+ self.assertFalse(is_ok)
+ self.assertTrue(any("File missing: misc.txt" in i for i in issues))
+
+ def test_manifest_versioning(self):
+ """Verify the manifest includes the new version and metadata fields."""
+ self.manager.save_manifest()
+
+ with open(self.manager.manifest_path) as f:
+ manifest = json.load(f)
+
+ self.assertEqual(manifest["version"], 2)
+ self.assertIn("metadata", manifest)
+
+ # Check if database metadata exists
+ db_rel = str(self.db_path.relative_to(self.test_dir))
+ self.assertIn(db_rel, manifest["metadata"])
+ self.assertIn("entropy", manifest["metadata"][db_rel])
+ self.assertIn("size", manifest["metadata"][db_rel])
+
+ def test_database_size_divergence(self):
+ """Verify size changes are caught when hash changes but entropy is similar."""
+ self.manager.save_manifest()
+
+ # Grow the database with similar content
+ conn = sqlite3.connect(self.db_path)
+ conn.execute("INSERT INTO data (val) VALUES (?)", ("more content" * 100,))
+ conn.commit()
+ conn.close()
+
+ is_ok, issues = self.manager.check_integrity()
+ # Even if entropy shift is low, hash and size changed
+ if not is_ok:
+ self.assertTrue(any("Database" in i for i in issues))
+
+
+if __name__ == "__main__":
+ unittest.main()
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────